/**
 * Copyright (c) 2010
 * Bert Young. UESTC. 
 */
 
#include "TCPTaskPool.h"

void VerifyThread::_add(TCPTask * task )
{
	//只监视可读，等待客户传来的验证消息
	task->epollAdd(this->epfd, EPOLLIN | EPOLLPRI, (void *)task );
	this->tasks.push_back(task);
	this->taskcount = tasks.size();
	if ( taskcount > events.size() )
		events.resize(taskcount + 16 );
#ifdef CHINESE
	debug("添加到验证线程：连接id %d", task->id);
#else
	debug("add to verify: taskid %d", task->id);
#endif
}

void VerifyThread::remove(std::list<TCPTask *>::iterator & iter)
{
#ifndef CHINESE
	debug("remove from verify: taskid %d", (*iter)->id);
#else
	debug("从验证线程移出：连接id %d", (*iter)->id);
#endif
	(*iter)->epollDel(this->epfd);
	iter = tasks.erase(iter);
	taskcount = tasks.size();
}

void VerifyThread::remove(TCPTask * task )
{
	task->epollDel(this->epfd);
	tasks.remove(task);//std::list::remove
	taskcount = tasks.size();
#ifndef CHINESE
	debug("remove from verify: taskid %d", task->id);
#else
	debug("从验证线程移出：连接id %d", task->id);
#endif
}

void VerifyThread::run( )
{
	Time current;
	std::list<TCPTask *>::iterator it;
	while( isAlive() )
	{
		current.now();
		checkQueue();//把main仍来的新连接收下,_add实现
		for ( it = tasks.begin(); it != tasks.end(); )
		{
			TCPTask * task = *it;
			/*这里可以检查是否change描述符，实现断线重连*/
			if ( task->checkVerifyTimeout(current))
			{
#ifdef CHINESE
			debug("连接超时，连接id = %d", task->id);
#else
			debug("verify timeout, task id = %d", task->id);
#endif
				//验证超时
				this->remove(it);
				task->resetState();
				pool->addRecycle(task);//bert ; will get next state
			}
			else
				++it;
		}

		int ret = epoll_wait(epfd, &events[0], taskcount, 0 );
		for (int i=0; i<ret; ++i)
		{
			TCPTask * task = (TCPTask*)events[i].data.ptr;
			if ( events[i].events & (EPOLLERR | EPOLLPRI) )
			{
				// socket error
#ifndef CHINESE
				debug("%s, error task id = %d",__PRETTY_FUNCTION__, task->id);
#else
				debug("验证线程中出错连接id = %d", task->id);
#endif
				this->remove(task);
				task->resetState();
				pool->addRecycle(task);
			}
			else if ( events[i].events & EPOLLIN)
			{
				switch(task->verifyConn() )
				{
					case 0:
					//polltimeout,下面会处理
						break;
					case 1://此时对端客户可能还没有从checkrebound返回呢
						this->remove(task);
						if(task->addToContainer())
						{
							pool->addOkay(task);
						}
						else
						{
							debug("the task id %d exist, impossible!", task->id);
							this->remove(task);// why not?
							task->resetState();
							pool->addRecycle(task);
						}
						break;
					case -1:
						this->remove(task);
						task->resetState();
						pool->addRecycle(task);
						break;
					}
				}
			}
		Thread::msleep(50);
	}

	for ( it = tasks.begin(); it!=tasks.end(); )
	{
		TCPTask * task = *it;
		this->remove(it);
		task->resetState();
		pool->addRecycle(task);
	}
}

//////////////////////////////////////////////////////////////////
void OkayThread::_add(TCPTask * task )
{
#ifdef CHINESE
	debug("添加到OKAY线程：连接id %d", task->id);
#else
	debug("add to okay: taskid %d", task->id);
#endif
	task->epollAdd(this->epfd, EPOLLIN | EPOLLOUT | EPOLLPRI, (void *)task );
	this->tasks.push_back(task);
	this->taskcount = tasks.size();
	if ( taskcount > events.size() )
		events.resize(taskcount + 16 );
	task->recv(false);
}

void OkayThread::remove(std::list<TCPTask *>::iterator & iter)
{
#ifdef CHINESE
	debug("从OKAY线程移出：连接id %d", (*iter)->id);
#else
	debug("remove from okay: taskid %d", (*iter)->id);
#endif
	(*iter)->epollDel(this->epfd);
	iter = tasks.erase(iter);
	this->taskcount = tasks.size();
}

void OkayThread::run( )
{
	Time current, write_time;
	std::list<TCPTask * >::iterator it;

	int epfd_read = epoll_create(256);
	std::vector<struct epoll_event> events_read;
	events_read.resize(256);
	unsigned int count_read = 0;
	bool check = false;

	while( isAlive() )
	{
		current.now();

		if ( check )
		{//将新连接加入只读监视epoll符;检测心跳；清洗无效连接
			check = false;
			checkQueue();//把main仍来的新连接收下
			for ( it = tasks.begin(); it != tasks.end(); )
			{
				TCPTask * task = *it;
				/*这里可以检查是否change描述符，实现断线重连*/
				if (!task->checkHeartbeat(current) )
					task->setDisconnect(TCPTask::disconnect_active);
				if ( task->shouldDisconnect())
				{
					if ( task->isEpollRead)
					{
						task->epollDel(epfd_read);
						task->isEpollRead = false;
						-- count_read;
					}
					this->remove(it);
					if (!task->removeFromContainer() )
						debug("task id %d is not in the container!", task->id);
					pool->addRecycle(task);
				}
				else
				{
					if ( !task->isEpollRead )
					{//往可读epoll监视符加入自己，可读计数加1
						task->epollAdd(epfd_read, EPOLLIN|EPOLLPRI, (void *)task);
						task->isEpollRead = true;
						++ count_read;
						if ( count_read > events_read.size() )
							events_read.resize(count_read+32);

					}
					++it;
				}
			}
		}//end if check

		Thread::msleep(2);

		//监视可读条件
		if ( count_read > 0 )
		{
			int ret = epoll_wait(epfd_read, &events_read[0], count_read, 0 );
			for (int i=0; i<ret; ++i)
			{
				TCPTask * task=(TCPTask*)events_read[i].data.ptr;
				if ( events_read[i].events & EPOLLERR )
				{
#ifndef CHINESE
					debug("error1 task id = %d", task->id);
#else
					debug("等待可读出错，连接 id = %d", task->id);
#endif
					if ( task->TerminateError() )
					{	//连接应断开，下一个循环处理它，见continue
						task->setDisconnect(TCPTask::disconnect_active);
						check = true;
					}
					else 
					{// 等待断线重连
						if ( task->isEpollRead )
						{
							task->epollDel(epfd_read);
							task->isEpollRead = false;
							-- count_read;
						}
						
						task->epollDel(epfd);
					}
				}
				else if ( events_read[i].events & EPOLLIN)
				{
					if ( !task->recv(true) )
					{
#ifndef CHINESE
						debug("error2 recv. task id = %d", task->id);
#else
						debug("接受数据出错，连接 id = %d", task->id);
#endif
						task->setDisconnect(TCPTask::disconnect_active);
						check = true;
					}
				}
				events_read[i].events = 0;
			}
		}// end if count_read
		
		if ( check )//有无效连接，回头处理去!
			continue;

		//50毫秒才检测写事件，避免从epoll模型来回移动
		if(current.msec()-write_time.msec() >= pool->usleep_time/1000)
		{
			write_time = current;
			int ret = epoll_wait(epfd, &events[0], taskcount, 0 );
			for (int i=0; i<ret; ++i)
			{
				TCPTask * task = (TCPTask * )events[i].data.ptr;
				if ( events[i].events & EPOLLERR )
				{
#ifndef CHINESE
					debug("error3 task id = %d", task->id);
#else
					debug("等待读写出错，连接 id = %d", task->id);
#endif
					if ( task->TerminateError() )
					{	//连接应断开，下一个循环处理它，见continue
						task->setDisconnect(TCPTask::disconnect_active);
					}
					else 
					{
						if ( task->isEpollRead )
						{
							task->epollDel(epfd_read);
							task->isEpollRead = false;
							-- count_read;
						}
					
						task->epollDel(epfd);
					}
				}
				else
				{
					if ( events[i].events & EPOLLIN)
					{
						if ( !task->recv(true) )
						{
#ifndef CHINESE
							debug("error4 recv. task id = %d", task->id);
#else
							debug("接受数据出错2，连接 id = %d", task->id);
#endif
							task->setDisconnect(TCPTask::disconnect_active);
						}
					}
					if ( events[i].events & EPOLLOUT)
					{
						if ( !task->send() )
						{
#ifndef CHINESE
							debug("error5 send. task id = %d", task->id);
#else
							debug("发送数据出错，连接 id = %d", task->id);
#endif
							task->setDisconnect(TCPTask::disconnect_active);
						}
					}
				}
				events_read[i].events = 0;
			}
			//发生了读写，可能出错，所以要检查 //50毫秒定时器都到了，也该检查了
			check = true;
		}//end if写事件
	}//end while alive

	for ( it = tasks.begin(); it!=tasks.end(); )
	{
		TCPTask * task = *it;
		this->remove(it);
		if (!task->removeFromContainer() )
			debug("task id %d is not in the container2!", task->id);
		pool->addRecycle(task);
	}

	TEMP_FAILURE_RETRY(::close(epfd_read));
}
//////////////////////////////////////////////
void RecycleThread::_add(TCPTask * task )
{
#ifdef CHINESE
	debug("添加到回收线程：连接id %d", task->id);
#else
	debug("add to recycle: taskid %d", task->id);
#endif
	this->tasks.push_front(task);
}
 
void RecycleThread::run( )
{
	std::list<TCPTask * >::iterator it;

	bool done = false;// 丑陋的做法，为了让所有连接得到回收
	while( isAlive() || !done )
	{
		done = !isAlive();
		checkQueue();
		for ( it = tasks.begin(); it != tasks.end(); )
		{
			TCPTask * task = *it;
			switch(task->recycleConn())
			{
				case 0:
					++it;
					break;
				case 1:
					it = tasks.erase(it);
#ifndef CHINESE
	debug("recycle taskid %d", task->id);
#else
	debug("回收连接id %d", task->id);
#endif
					//task->nextState();
					delete task;
					task = NULL;
					break;
			}
		}
		Thread::msleep(200);
	}
}

///////////////////////////////////////////////
///////////////////////////////////////////////
long TCPTaskPool::usleep_time = 50000;

bool TCPTaskPool::addVerify(TCPTask * task)
{
	VerifyThread * pthread = verifyThreads.getOne();
	if ( pthread != NULL )
	{
		task->nextState();
		pthread->addTask(task);
		return true;
	}
	else
	{
		debug("my god, cannot find verify thread");
		return false;
	}
}

bool TCPTaskPool::addOkay(TCPTask * task)
{
	OkayThread * pthread = okayThreads.getOne();
	if ( pthread != NULL )
	{
		task->nextState();
		pthread->addTask(task);
		return true;
	}
	else
	{
		debug("my god, cannot find okay thread");
		return false;
	}
}

void TCPTaskPool::addRecycle(TCPTask * task)
{
	RecycleThread * pthread = recycleThreads.getOne();
	if ( pthread != NULL )
	{
		task->resetState();
		pthread->addTask(task);
	}
	else
		debug("my god, cannot find recycle thread");
}

bool TCPTaskPool::init()
{
	if ( !recycleThreads.init(1, 1, "回收线程", this))
		return false;
	if ( !verifyThreads.init(1, 1, "验证线程", this))
		return false;
	int maxThread = (maxConns + OkayThread::max_task_size -1 ) / OkayThread::max_task_size;//4
	if ( !okayThreads.init(1, maxThread, "Okay线程", this))
		return false;
	return true;
}

void TCPTaskPool::finally()
{
	//主服务器重写baseserver的recycle函数，调用此finally
	verifyThreads.stopAllThreads();
	okayThreads.stopAllThreads();
	recycleThreads.stopAllThreads();
}
